package com.atguigu.flink.datastreamapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/4
 *
 * process算子是最底层的算子。
 *  map,filter,flatmap这些算子的功能都可以由process实现
 */
public class Demo2_ProcessSum
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                                                               .map(new WaterSensorMapFunction());

        //模拟process 进行所有传感器的聚合
        ds
            //.global()
            .process(new ProcessFunction<WaterSensor, String>()
        {

            //属性，属于当前对象
            int sumVc = 0;
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                sumVc += value.getVc();

                out.collect("最新的vc之和:"+sumVc);

            }
        }).setParallelism(1)
          .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
